initial commit for prototype
This commit is contained in:
shell committed
commit
2c1bac9bcb
14 files changed
+883
No files matched your search
@@ -0,0 +1,8 @@
|
||||
from mini_agent.react_agent import ReActAgent
|
||||
from mini_agent.tools.tool_manager import ToolManager
|
||||
|
||||
if __name__ == "__main__":
|
||||
agent = ReActAgent(tool_manager=ToolManager())
|
||||
while True:
|
||||
question = input("提问：")
|
||||
print(agent.run(question))
|
||||
Whitespace-only changes.
@@ -0,0 +1,153 @@
|
||||
import json
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from mini_agent.tools.tavily_search import TavilySearch
|
||||
from openai import OpenAI
|
||||
|
||||
from mini_agent.tools.tool_search import ToolSearch
|
||||
from mini_agent.tools.tool_manager import ToolManager
|
||||
from dotenv import load_dotenv
|
||||
|
||||
load_dotenv()
|
||||
|
||||
class ReActAgent:
|
||||
SYSTEM_PROMPT = """
|
||||
你是一个能够调用外部工具的智能助手。
|
||||
|
||||
工作规则：
|
||||
|
||||
1. 首先理解用户的问题，判断是否需要外部工具。
|
||||
2. 如果需要工具，调用 search_tools 搜索合适的工具。
|
||||
3. 根据搜索结果选择合适的工具。
|
||||
4. 工具执行后，根据返回结果判断下一步行动。
|
||||
5. 如果需要其他工具，可以再次调用 search_tools。
|
||||
6. 当已经获得足够的信息，能够完整回答用户问题时，直接给出最终答案。
|
||||
|
||||
注意事项：
|
||||
|
||||
- 不允许编造工具执行结果。
|
||||
- 不允许调用尚未通过 search_tools 发现的业务工具。
|
||||
- 工具搜索没有结果时，可以尝试调整关键词重新搜索。
|
||||
- 如果工具无法完成任务，明确告知用户。
|
||||
- 不要为了调用工具而调用工具。
|
||||
- 只有在任务已经完成或明确无法继续时才结束回答。
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tool_manager: ToolManager,
|
||||
model: str | None = None,
|
||||
api_key: str | None = None,
|
||||
base_url: str | None = None,
|
||||
timeout: int | None = None,
|
||||
max_steps: int = 10,
|
||||
max_tool_calls: int = 10,
|
||||
) -> None:
|
||||
|
||||
self.model = model or os.getenv("LLM_MODEL_ID")
|
||||
self.base_url = base_url or os.getenv("LLM_BASE_URL")
|
||||
self.api_key = api_key or os.getenv("LLM_API_KEY")
|
||||
self.timeout = timeout or int(os.getenv("LLM_TIMEOUT", 60))
|
||||
|
||||
if not all([self.model, self.base_url, self.api_key]):
|
||||
raise ValueError(
|
||||
"模型ID、API密钥和服务地址必须被提供或在.env文件中定义。"
|
||||
)
|
||||
|
||||
self.client = OpenAI(
|
||||
api_key=self.api_key,
|
||||
base_url=self.base_url,
|
||||
timeout=self.timeout,
|
||||
)
|
||||
|
||||
self.tool_manager = tool_manager
|
||||
self.max_steps = max_steps
|
||||
self.max_tool_calls = max_tool_calls
|
||||
self.search_tool = ToolSearch(self.tool_manager)
|
||||
self.tool_manager.register(self.search_tool)
|
||||
web_search_tool = TavilySearch()
|
||||
web_search_tool.config(**{"api_key":os.getenv("TAVILY_KEY","")})
|
||||
self.tool_manager.register(web_search_tool)
|
||||
|
||||
def run(self, question: str) -> str:
|
||||
|
||||
# 已发现的业务工具
|
||||
|
||||
# 初始只开放工具发现能力
|
||||
available_tools = [self.search_tool]
|
||||
tool_calls_count = 0
|
||||
|
||||
messages = [
|
||||
{"role":"user","content":question}
|
||||
]
|
||||
|
||||
for step in range(self.max_steps):
|
||||
|
||||
print(f"\n========== Step {step + 1} ==========")
|
||||
|
||||
tools = [self.tool_manager.tool_to_openai(tool) for tool in available_tools]
|
||||
|
||||
response = self.client.chat.completions.create(
|
||||
model=self.model,
|
||||
messages=messages,
|
||||
tools=tools,
|
||||
tool_choice="auto",
|
||||
)
|
||||
|
||||
message = response.choices[0].message
|
||||
|
||||
# 模型没有调用工具，说明它选择直接回答
|
||||
if not message.tool_calls:
|
||||
if message.content:
|
||||
print(f"最终回答: {message.content}")
|
||||
return message.content
|
||||
raise RuntimeError("模型没有返回有效内容")
|
||||
|
||||
# 保存模型的工具调用消息
|
||||
messages.append(
|
||||
message.model_dump(exclude_none=True)
|
||||
)
|
||||
|
||||
for tool_call in message.tool_calls:
|
||||
tool_name = tool_call.function.name
|
||||
try:
|
||||
arguments = json.loads(
|
||||
tool_call.function.arguments or "{}"
|
||||
)
|
||||
except json.JSONDecodeError:
|
||||
observation = json.dumps(
|
||||
{"error": "工具参数不是合法的 JSON"},
|
||||
ensure_ascii=False,
|
||||
)
|
||||
|
||||
messages.append({
|
||||
"role": "tool",
|
||||
"tool_call_id": tool_call.id,
|
||||
"content": observation,
|
||||
})
|
||||
continue
|
||||
|
||||
print(f"调用工具: {tool_name}")
|
||||
print(f"参数: {arguments}")
|
||||
|
||||
|
||||
observation = self.tool_manager.call_tool(tool_name,arguments)
|
||||
# 更新当前可用工具
|
||||
if tool_name == self.search_tool.name:
|
||||
new_tools = []
|
||||
for item in observation:
|
||||
if not any(tool.name == item.name for tool in available_tools):
|
||||
available_tools.append(item)
|
||||
new_tools.append(item)
|
||||
observation = [self.tool_manager.tool_to_openai(tool) for tool in new_tools]
|
||||
|
||||
messages.append({
|
||||
"role": "tool",
|
||||
"tool_call_id": tool_call.id,
|
||||
"content": str(observation),
|
||||
})
|
||||
|
||||
raise RuntimeError(
|
||||
f"Agent 已达到最大执行步数 ({self.max_steps})，任务未完成"
|
||||
)
|
||||
Whitespace-only changes.
@@ -0,0 +1,60 @@
|
||||
from mini_agent.tools.tool import Configuration, Tool,Parameter
|
||||
from pydantic import BaseModel, Field
|
||||
from tavily import TavilyClient
|
||||
|
||||
class ResultItem(BaseModel):
|
||||
url:str
|
||||
title:str
|
||||
content:str
|
||||
score:float
|
||||
raw_content:str | None = None
|
||||
id:str
|
||||
|
||||
|
||||
class SearchResult(BaseModel):
|
||||
query:str
|
||||
answer:str | None
|
||||
images:list[str] = []
|
||||
results: list[ResultItem]
|
||||
|
||||
class TavilySearch(Tool[SearchResult]):
|
||||
|
||||
@property
|
||||
def name(self)->str:
|
||||
return "Tavily网络搜索工具 "
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return """
|
||||
使用 Tavily 搜索引擎进行互联网搜索，获取最新网页信息。
|
||||
适用于查询时效性信息、新闻、技术文档、产品资料以及需要外部信息验证的问题。
|
||||
"""
|
||||
|
||||
@property
|
||||
def catalog(self) -> str:
|
||||
return "web.search"
|
||||
|
||||
@property
|
||||
def configurations(self) -> list[Configuration]:
|
||||
return [
|
||||
Configuration(name="api_key",type="str",required=True)
|
||||
]
|
||||
|
||||
@property
|
||||
def parameters(self) -> list[Parameter]:
|
||||
return [
|
||||
Parameter(name="query",type="str",description="query keyword",required=True)
|
||||
]
|
||||
|
||||
def config(self,**kwargs) -> None:
|
||||
api_key = kwargs.get("api_key",None)
|
||||
if not api_key:
|
||||
raise ValueError("API Key is required")
|
||||
self.client = TavilyClient(api_key=api_key)
|
||||
|
||||
def run(self, **kwargs) -> SearchResult:
|
||||
query = kwargs.get("query",None)
|
||||
if not query:
|
||||
raise ValueError("查询参数不能为空")
|
||||
response = self.client.search(query=query,search_depth="advanced",include_answer="advanced")
|
||||
return SearchResult.model_validate(response)
|
||||
@@ -0,0 +1,49 @@
|
||||
from pydantic import BaseModel
|
||||
from abc import ABC,abstractmethod
|
||||
|
||||
|
||||
class Parameter(BaseModel):
|
||||
name:str
|
||||
type:str
|
||||
description:str
|
||||
required:bool =True
|
||||
|
||||
class Configuration(BaseModel):
|
||||
name:str
|
||||
type:str
|
||||
required:bool = True
|
||||
|
||||
class Tool[T](ABC):
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def name(self)->str:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def description(self)->str:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def catalog(self)->str:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def parameters(self)->list[Parameter]:
|
||||
pass
|
||||
|
||||
@property
|
||||
@abstractmethod
|
||||
def configurations(self)->list[Configuration]:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def config(self,**kwargs)->None:
|
||||
pass
|
||||
|
||||
@abstractmethod
|
||||
def run(self,**kwargs)->T:
|
||||
pass
|
||||
@@ -0,0 +1,80 @@
|
||||
from typing import Any
|
||||
|
||||
from mini_agent.tools.tool import Tool
|
||||
|
||||
|
||||
class ToolManager:
|
||||
|
||||
_tools:dict[str,Tool] = {}
|
||||
|
||||
def register[T](self,tool:Tool[T])->None:
|
||||
if tool.name in self._tools:
|
||||
raise ValueError(f"{tool.name} is already registered")
|
||||
self._tools[tool.name] = tool
|
||||
|
||||
def get_tools(self,keyword:str,catalog:str | None)->list[Tool]:
|
||||
return list(self._tools.values())
|
||||
|
||||
def get_tools_for_openai(self,keyword:str,catalog:str|None)->list[dict[str,Any]]:
|
||||
tools = self.get_tools(keyword,catalog)
|
||||
result = []
|
||||
for tool in tools:
|
||||
result.append(self.tool_to_openai(tool))
|
||||
return result
|
||||
|
||||
def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
|
||||
tool = self._tools.get(name, None)
|
||||
|
||||
if tool is None:
|
||||
raise ValueError(f"Tool not found: {name}")
|
||||
|
||||
return tool.run(**arguments)
|
||||
|
||||
def tool_to_openai[T](self,tool:Tool[T])->dict[str,Any]:
|
||||
type_mapping = {
|
||||
"str": "string",
|
||||
"string": "string",
|
||||
"int": "integer",
|
||||
"integer": "integer",
|
||||
"float": "number",
|
||||
"number": "number",
|
||||
"bool": "boolean",
|
||||
"boolean": "boolean",
|
||||
"list": "array",
|
||||
"dict": "object",
|
||||
}
|
||||
properties = {}
|
||||
required = []
|
||||
for parameter in tool.parameters:
|
||||
json_type = type_mapping.get(parameter.type)
|
||||
|
||||
if json_type is None:
|
||||
raise ValueError(
|
||||
f"不支持的参数类型: {parameter.type}"
|
||||
)
|
||||
|
||||
properties[parameter.name] = {
|
||||
"type": json_type,
|
||||
}
|
||||
|
||||
if parameter.description:
|
||||
properties[parameter.name]["description"] = (
|
||||
parameter.description
|
||||
)
|
||||
|
||||
if parameter.required:
|
||||
required.append(parameter.name)
|
||||
|
||||
return {
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": tool.name,
|
||||
"description": tool.description.strip(),
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": properties,
|
||||
"required": required,
|
||||
"additionalProperties": False,
|
||||
},
|
||||
},
|
||||
}
|
||||
@@ -0,0 +1,43 @@
|
||||
from mini_agent.tools.tool import Tool,Parameter,Configuration
|
||||
from mini_agent.tools.tool_manager import ToolManager
|
||||
|
||||
class ToolSearch(Tool[list[Tool]]):
|
||||
|
||||
def __init__(self,tool_manager:ToolManager) -> None:
|
||||
self.tool_manager = tool_manager
|
||||
|
||||
@property
|
||||
def name(self)->str:
|
||||
return """search_tools"""
|
||||
|
||||
@property
|
||||
def description(self) -> str:
|
||||
return """
|
||||
"根据用户需求搜索可用工具。 当需要外部数据、查询业务信息或执行特定操作时，首先使用该工具查找合适的业务工具。可以根据需要多次搜索不同的工具。
|
||||
"""
|
||||
|
||||
@property
|
||||
def catalog(self) -> str:
|
||||
return "system"
|
||||
|
||||
def config(self, **kwargs) -> None:
|
||||
pass
|
||||
|
||||
@property
|
||||
def configurations(self) -> list[Configuration]:
|
||||
return []
|
||||
|
||||
@property
|
||||
def parameters(self) -> list[Parameter]:
|
||||
return [
|
||||
Parameter(name="keyword",type="str",description="关键词",required=True),
|
||||
Parameter(name="catalog",type="str",description="工具分类",required=False)
|
||||
]
|
||||
|
||||
def run(self, **kwargs) -> list[Tool]:
|
||||
keyword = kwargs.get("keyword", None)
|
||||
catalog = kwargs.get("catalog", None)
|
||||
if not keyword:
|
||||
raise ValueError("keyword is required for query")
|
||||
|
||||
return self.tool_manager.get_tools(keyword,catalog)
|
||||
Reference in new issue
Block a user