400 lines
12 KiB
Python
400 lines
12 KiB
Python
"""
|
||
LLM客户端 - 调用大模型API
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
import time
|
||
from typing import Dict, List, Optional, Any
|
||
|
||
|
||
class LLMClient:
|
||
"""大模型客户端"""
|
||
|
||
def __init__(
|
||
self,
|
||
base_url: str = "http://192.168.2.17:19007/v1",
|
||
api_key: str = "xxxx",
|
||
model: str = "auto",
|
||
timeout: int = 120
|
||
):
|
||
self.base_url = base_url.rstrip('/')
|
||
self.api_key = api_key
|
||
self.model = model
|
||
self.timeout = timeout
|
||
|
||
def chat(
|
||
self,
|
||
messages: List[Dict],
|
||
temperature: float = 0.7,
|
||
max_tokens: int = 4096,
|
||
stream: bool = False
|
||
) -> Dict:
|
||
"""
|
||
发送聊天请求
|
||
|
||
Args:
|
||
messages: [{"role": "user/assistant/system", "content": "..."}]
|
||
temperature: 温度参数
|
||
max_tokens: 最大token数
|
||
stream: 是否流式输出
|
||
|
||
Returns:
|
||
{"content": "...", "usage": {...}}
|
||
"""
|
||
url = f"{self.base_url}/chat/completions"
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {self.api_key}"
|
||
}
|
||
|
||
payload = {
|
||
"model": self.model,
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
"max_tokens": max_tokens,
|
||
"stream": stream
|
||
}
|
||
|
||
try:
|
||
response = requests.post(
|
||
url,
|
||
headers=headers,
|
||
json=payload,
|
||
timeout=self.timeout,
|
||
stream=stream
|
||
)
|
||
|
||
if response.status_code != 200:
|
||
return {
|
||
"content": "",
|
||
"error": f"API错误: {response.status_code} - {response.text}"
|
||
}
|
||
|
||
if stream:
|
||
# 流式处理
|
||
content = ""
|
||
for line in response.iter_lines():
|
||
if line:
|
||
line = line.decode('utf-8')
|
||
if line.startswith('data: '):
|
||
data = line[6:]
|
||
if data == '[DONE]':
|
||
break
|
||
try:
|
||
chunk = json.loads(data)
|
||
if 'choices' in chunk and len(chunk['choices']) > 0:
|
||
delta = chunk['choices'][0].get('delta', {})
|
||
if 'content' in delta:
|
||
content += delta['content']
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
return {"content": content}
|
||
|
||
else:
|
||
data = response.json()
|
||
content = data['choices'][0]['message']['content']
|
||
usage = data.get('usage', {})
|
||
|
||
return {
|
||
"content": content,
|
||
"usage": usage
|
||
}
|
||
|
||
except requests.Timeout:
|
||
return {"content": "", "error": "请求超时"}
|
||
except requests.RequestException as e:
|
||
return {"content": "", "error": f"请求异常: {str(e)}"}
|
||
|
||
def simple_chat(self, prompt: str, system_prompt: str = "") -> str:
|
||
"""
|
||
简单聊天接口
|
||
|
||
Args:
|
||
prompt: 用户输入
|
||
system_prompt: 系统提示
|
||
|
||
Returns:
|
||
模型回复文本
|
||
"""
|
||
messages = []
|
||
if system_prompt:
|
||
messages.append({"role": "system", "content": system_prompt})
|
||
messages.append({"role": "user", "content": prompt})
|
||
|
||
result = self.chat(messages)
|
||
|
||
if "error" in result and result["error"]:
|
||
raise Exception(result["error"])
|
||
|
||
return result["content"]
|
||
|
||
def structured_output(
|
||
self,
|
||
prompt: str,
|
||
schema: Dict,
|
||
system_prompt: str = ""
|
||
) -> Dict:
|
||
"""
|
||
结构化输出
|
||
|
||
Args:
|
||
prompt: 用户输入
|
||
schema: 输出格式描述
|
||
system_prompt: 系统提示
|
||
|
||
Returns:
|
||
解析后的JSON对象
|
||
"""
|
||
schema_text = json.dumps(schema, indent=2, ensure_ascii=False)
|
||
|
||
full_system = system_prompt + "\n\n请按照以下JSON格式输出，不要输出其他内容:\n" + schema_text
|
||
|
||
messages = [
|
||
{"role": "system", "content": full_system},
|
||
{"role": "user", "content": prompt}
|
||
]
|
||
|
||
result = self.chat(messages, temperature=0.3)
|
||
|
||
if "error" in result and result["error"]:
|
||
raise Exception(result["error"])
|
||
|
||
content = result["content"].strip()
|
||
|
||
# 尝试解析JSON
|
||
try:
|
||
# 去除可能的markdown代码块标记
|
||
if content.startswith('```'):
|
||
lines = content.split('\n')
|
||
content = '\n'.join(lines[1:-1] if lines[-1] == '```' else lines[1:])
|
||
|
||
return json.loads(content)
|
||
except json.JSONDecodeError:
|
||
# 尝试提取JSON部分
|
||
import re
|
||
json_match = re.search(r'\{.*\}', content, re.DOTALL)
|
||
if json_match:
|
||
try:
|
||
return json.loads(json_match.group())
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
return {"raw_content": content, "parse_error": "无法解析为JSON"}
|
||
|
||
def analyze_intent(self, user_request: str) -> Dict:
|
||
"""
|
||
分析用户意图
|
||
|
||
Args:
|
||
user_request: 用户原始请求
|
||
|
||
Returns:
|
||
{"intent": "...", "keywords": [...], "need_clarification": bool, "questions": [...]}
|
||
"""
|
||
system_prompt = """你是一个意图分析专家。分析用户的请求，判断：
|
||
1. 用户的核心意图是什么
|
||
2. 提取关键信息
|
||
3. 信息是否足够完整（不需要额外澄清）
|
||
4. 如果不完整，需要澄清的问题
|
||
|
||
请以JSON格式输出。"""
|
||
|
||
schema = {
|
||
"intent": "用户核心意图的简洁描述",
|
||
"keywords": ["关键信息列表"],
|
||
"need_clarification": "是否需要澄清 (true/false)",
|
||
"questions": ["需要澄清的问题列表（如果need_clarification为true）"]
|
||
}
|
||
|
||
return self.structured_output(
|
||
f"分析以下用户请求:\n\n{user_request}",
|
||
schema,
|
||
system_prompt
|
||
)
|
||
|
||
def split_tasks(self, intent: Dict, user_request: str) -> List[Dict]:
|
||
"""
|
||
任务拆分
|
||
|
||
Args:
|
||
intent: 意图分析结果
|
||
user_request: 用户原始请求
|
||
|
||
Returns:
|
||
[{"id": "...", "type": "serial/parallel", "description": "...", "dependencies": [...]}]
|
||
"""
|
||
system_prompt = """你是一个任务规划专家。根据用户意图，将请求拆分为多个子任务。
|
||
|
||
规则：
|
||
1. 识别任务之间的依赖关系
|
||
2. 无依赖的任务标记为parallel（可并行）
|
||
3. 有依赖的任务标记为serial（串行）
|
||
4. 每个任务有明确的描述
|
||
5. 每个任务有明确的输入输出要求
|
||
|
||
请以JSON格式输出任务列表。"""
|
||
|
||
schema = {
|
||
"tasks": [
|
||
{
|
||
"id": "任务唯一标识（如task_1, task_2）",
|
||
"type": "serial 或 parallel",
|
||
"description": "任务描述",
|
||
"input_schema": {"输入要求"},
|
||
"output_schema": {"输出要求"},
|
||
"dependencies": ["依赖的任务ID列表"]
|
||
}
|
||
],
|
||
"execution_order": ["任务执行顺序说明"]
|
||
}
|
||
|
||
result = self.structured_output(
|
||
f"""用户原始请求: {user_request}
|
||
|
||
意图分析结果:
|
||
{json.dumps(intent, indent=2, ensure_ascii=False)}
|
||
|
||
请拆分为子任务列表。""",
|
||
schema,
|
||
system_prompt
|
||
)
|
||
|
||
return result.get("tasks", [])
|
||
|
||
def generate_bid(
|
||
self,
|
||
task: Dict,
|
||
agent_profile: Dict
|
||
) -> Dict:
|
||
"""
|
||
Agent生成竞标
|
||
|
||
Args:
|
||
task: 任务定义
|
||
agent_profile: Agent档案
|
||
|
||
Returns:
|
||
竞标书内容
|
||
"""
|
||
system_prompt = """你是一个执行Agent，需要为任务竞标。
|
||
|
||
评估你的能力和任务的匹配度，给出：
|
||
1. 能力匹配度（0-1）
|
||
2. 预估完成时间（秒）
|
||
3. 自信度（0-1）
|
||
4. 执行方案描述
|
||
5. 前置条件（如果有）
|
||
6. 备选方案（如果有）
|
||
|
||
请以JSON格式输出。"""
|
||
|
||
schema = {
|
||
"capability_match": "能力匹配度 0-1",
|
||
"estimated_time": "预估完成时间(秒)",
|
||
"confidence": "自信度 0-1",
|
||
"approach": "执行方案描述",
|
||
"prerequisites": ["前置条件列表"],
|
||
"alternative_approaches": ["备选方案列表"]
|
||
}
|
||
|
||
result = self.structured_output(
|
||
f"""任务:
|
||
{json.dumps(task, indent=2, ensure_ascii=False)}
|
||
|
||
你的档案:
|
||
{json.dumps(agent_profile, indent=2, ensure_ascii=False)}
|
||
|
||
请生成竞标书。""",
|
||
schema,
|
||
system_prompt
|
||
)
|
||
|
||
return result
|
||
|
||
def execute_task(
|
||
self,
|
||
task: Dict,
|
||
approach: str = ""
|
||
) -> Dict:
|
||
"""
|
||
执行任务
|
||
|
||
Args:
|
||
task: 任务定义
|
||
approach: 执行方案
|
||
|
||
Returns:
|
||
执行结果
|
||
"""
|
||
system_prompt = """你是一个任务执行者。根据任务描述和执行方案，完成任务并输出结果。
|
||
|
||
输出应该符合任务的output_schema要求。"""
|
||
|
||
messages = [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": f"""任务:
|
||
{json.dumps(task, indent=2, ensure_ascii=False)}
|
||
|
||
执行方案: {approach}
|
||
|
||
请执行任务并输出结果。"""}
|
||
]
|
||
|
||
result = self.chat(messages, temperature=0.5, max_tokens=8192)
|
||
|
||
if "error" in result and result["error"]:
|
||
return {"error": result["error"]}
|
||
|
||
return {"result": result["content"]}
|
||
|
||
def validate_result(
|
||
self,
|
||
task: Dict,
|
||
result: Any
|
||
) -> Dict:
|
||
"""
|
||
验证结果
|
||
|
||
Args:
|
||
task: 任务定义
|
||
result: 执行结果
|
||
|
||
Returns:
|
||
{"passed": bool, "issues": [...], "score": 0-1}
|
||
"""
|
||
system_prompt = """你是一个结果验证专家。评估执行结果是否符合任务要求。
|
||
|
||
判断：
|
||
1. 结果完整性（是否包含所有必要部分）
|
||
2. 结果正确性（是否符合预期）
|
||
3. 结果质量评分（0-1）
|
||
|
||
请以JSON格式输出。"""
|
||
|
||
schema = {
|
||
"passed": "是否通过验证 (true/false)",
|
||
"completeness": "完整性检查结果",
|
||
"correctness": "正确性检查结果",
|
||
"issues": ["问题列表（如果未通过）"],
|
||
"score": "质量评分 0-1",
|
||
"suggestions": ["改进建议"]
|
||
}
|
||
|
||
return self.structured_output(
|
||
f"""任务:
|
||
{json.dumps(task, indent=2, ensure_ascii=False)}
|
||
|
||
执行结果:
|
||
{json.dumps(result, indent=2, ensure_ascii=False) if isinstance(result, dict) else str(result)}
|
||
|
||
请验证结果。""",
|
||
schema,
|
||
system_prompt
|
||
)
|
||
|
||
|
||
# 默认客户端实例
|
||
default_client = LLMClient() |