業務邏輯編排對比
首先回顧一下我們的業務流程：
接下來分別使用手寫代碼和幾種框架來實現這個業務流程邏輯，看看差異所在。
1、手寫代碼
純編碼時，我們直接根據業務邏輯來串聯工作流：
# sql agent
def sql_agent(user_query):
# 1. 語義匹配
table = semanticService.hybrid_search(user_query, 1)
if not table:
print(f"未匹配到字段")
return Result.error()
table_struct = [t["table_info"] for t in table]
prompt = f"""
你是一個MySQL專家。根據以下表結構信息：
{table_struct}
歷史問答（僅供參考）："{mm.get_messages(session_id)}"
用户查詢："{user_query}"
生成標準MYSQL查詢語句。
要求：
1. 只輸出MYSQL語句，不要額外解釋
2. 根據語義和字段類型，使用COUNT/SUM/AVG等聚合函數進行計算，非必須
3. 給生成的字段取一個簡短的中文名稱
輸出格式：使用[]包含sql文本即可，不需要其他輸出，便於解析，例如:[select 1 from dual]
"""
print(f"SQL AGENT PROMPT={prompt}")
# 2. 大模型生成SQL
str1 = analysisService.analysis(prompt)
sql = re.search(r'\[(.*?)\]', str1, re.DOTALL).group(1).strip()
# 3. 執行查詢
if not sql:
print("\nSQL生成失敗")
return Result.error()
resultSet = queryService.query_with_column(sql)
if not resultSet:
print("\nSQL查詢失敗")
return Result.error()
return Result.success(data={ "tableStruct": table_struct, "resultSet": resultSet, "sql": sql })
# analysis agent
def analysis_agent(user_query, data):
# 基礎分析
prompt = f"""
根據以下表結構信息：
{data['tableStruct']}
查詢SQL：
{data['sql']}
和以下數據信息：
{data['resultSet']}
歷史問答（僅供參考）："{mm.get_messages(session_id)}"
用户查詢："{user_query}"
生成一段簡要分析，加上一些預測總結的內容
"""
print(f"ANALYSIS AGENT PROMPT={prompt}")
return Result.success(analysisService.analysis(prompt))
def workflow(user_input):
# 1 - SQL Agent
result = sql_agent(user_input)
if not result.success:
returnNone
# 2 - Analysis Agent
return analysis_agent(user_input, result.data)2、LangChain框架
使用LangChain框架的Chain模式時，可以手動串聯業務流程邏輯：
# 構建順序鏈
overall_chain = SequentialChain(
chains=[
TransformChain(
input_variables=["user_query"],
output_variables=["table_schema"],
transform=get_table_schema
),
TransformChain(
input_variables=["user_query", "table_schema"],
output_variables=["sql_result", "generated_sql"],
transform=execute_sql
),
analysis_chain
],
input_variables=["user_query"],
output_variables=["analysis_result"],
verbose=True
)LangChain 已經發展出豐富的 Chain 類型，用於構建複雜、模塊化的 LLM 應用，這裏就不一一介紹了。
LangChain框架除了提供Chain模式外，還可以使用的Agent模式來實現，通過提示詞來指定業務流程邏輯，然後直接調用工具，如下：
# 初始化Agent
agent = initialize_agent(
tools=[semantic_tool, sql_tool],
llm=llm,
agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
verbose=True,
handle_parsing_errors=True,
max_iteratinotallow=10, # 增加最大迭代次數以支持多步查詢
early_stopping_method="generate"
)
def chat(user_query):
print("執行方法chat")
# 更明確的指令
enhanced_query = f"""
問題：{user_query}
請特別注意：這個問題可能需要從多個表中查詢數據。
1. 首先確定需要查詢哪些數據
2. 使用match_metadata工具分別匹配包含這些數據的表結構
3. 對每個表生成相應的SQL查詢語句
4. 執行查詢並彙總結果
5. 最後計算並給出答案
請確保逐步執行，不要跳過任何步驟。
"""
try:
result = agent.run(enhanced_query)
print(f"\n最終結果: {result}")
except Exception as e:
print(f"執行過程中出錯: {str(e)}")
# 這裏可以添加重試或更詳細的錯誤處理邏輯3、QwenAgent框架
QwenAgent框架則只能通過提示詞方式指定業務流程邏輯，然後直接調用Assistant等待用户輸入：
# 創建Agent實例
agent = Assistant(
name='ai_agent_assistant',
llm={
'model': 'qwen3:32b',
'model_server': 'http://localhost:11434/v1',
},
system_message="""
你是一個數據分析助手，負責幫助用户查詢數據庫信息。
請特別注意：用户的問題可能需要從多個表中查詢數據。
1. 首先確定需要查詢哪些數據
2. 使用match_metadata工具分別匹配包含這些數據的表結構
3. 對每個表生成相應的SQL查詢語句
4. 執行查詢並彙總結果
5. 最後計算並給出答案
請確保逐步執行，不要跳過任何步驟。
""",
function_list=[MatchMetadataTool(), ExecuteSQLTool()],
)4、AgentScope框架
AgentScope框架中支持單體Agent模式，將工具提供給框架，再通過提示詞來指定業務流程邏輯：
async def interactive_react_agent() -> None:
"""創建一個支持多輪對話的ReAct智能體。"""
# 準備工具
toolkit = Toolkit()
toolkit.register_tool_function(match_metadata)
toolkit.register_tool_function(execute_sql)
jarvis = ReActAgent(
name="Jarvis",
sys_prompt="""
你是一個數據分析助手，負責幫助用户查詢數據庫信息。
請特別注意：用户的問題可能需要從多個表中查詢數據。
1. 首先確定需要查詢哪些數據
2. 使用match_metadata工具分別匹配包含這些數據的表結構
3. 對每個表生成相應的SQL查詢語句
4. 執行查詢並彙總結果
5. 最後計算並給出答案
請確保逐步執行，不要跳過任何步驟。
""",
model=OllamaChatModel(
model_name="qwen3:32b", # 指定模型名稱
stream=True, # 根據需要設置是否流式輸出
enable_thinking=True, # 為Qwen3啓用思考功能（可選）
# host="http://localhost:11434" # 如果Ollama不在默認地址，需指定
),
formatter=OllamaChatFormatter(),
toolkit=toolkit,
memory=InMemoryMemory(),
)小結
手寫代碼需人工串聯各個步驟，靈活但開發成本高，維護複雜。各框架則提供了豐富的鏈（Chain）或Agent模式，支持流程模塊化、工具調用自動化，極大提升開發效率和可維護性。
模型調用對比
接下來看看手寫代碼和使用開發框架，都是如何調用大模型的，如何處理大模型的流式輸出的。
1、手寫代碼
通過HTTP方式調用外部大模型示例：
import requests
# Analysis API
class AnalysisService:
def __init__(self):
self.ollama_host = "http://localhost:11434/api/chat"
def analysis(self, prompt, model="deepseek-r1:32b", messages=None):
# 發送POST請求
if messages isNone:
messages = []
str = ""
newMessages = messages[:]
newMessages.append({"role": "user", "content": prompt})
# 請求數據
data = {
"model": model,
"messages": newMessages,
"stream": True
}
isThinking = False
with requests.post(self.ollama_host, jsnotallow=data, stream=True) as response:
# 處理流式響應
for line in response.iter_lines():
if line:
decoded_line = line.decode('utf-8')
try:
# 解析JSON數據
chunk = json.loads(decoded_line)
content = chunk['message']['content']
if"<think>"in content:
isThinking = True
if"</think>"in content:
isThinking = False
ifnot isThinking and"</think>"notin content:
str += chunk['message']['content']
# 打印消息內容
print(chunk['message']['content'], end='', flush=True)
except json.JSONDecodeError:
print(f"無法解析JSON: {decoded_line}")
return str使用官方或社區封裝的 Python 包（如 ollama 包）調用示例：
import ollama
# 調用模型生成回覆，流式輸出
for chunk in ollama.chat(
model='deepseek-r1:32b',
messages=[{'role': 'user', 'content': '講一個笑話'}],
stream=True
):
print(chunk['message']['content'], end='', flush=True)2、LangChain框架
LangChain內部集成支持多種類型的模型調用，以下列舉部分：
3、QwenAgent框架
QwenAgent內部集成支持通用HTTP方式調用大模型：
...
# 創建Agent實例
agent = Assistant(
name='ai_agent_assistant',
llm={
'model': 'qwen3:32b',
'model_server': 'http://localhost:11434/v1',
},
...4、AgentScope框架
AgentScope內部集成多模型支持：
小結
手寫代碼通常直接調用HTTP接口或使用官方SDK，需自行處理流式輸出的循環響應和異常。而幾種開發框架都內置對多種模型的統一封裝，只要簡單配置即可，簡化了調用過程，提升了穩定性和擴展性。
工具調用對比
現在AI Agent中調用工具Tools已經是典型場景了，所以接下來對這塊做個對比分析。
1、手寫代碼
需要手動調用外部工具：
...
# sql agent
def sql_agent(user_query):
# 1. 語義匹配
table = semanticService.hybrid_search(user_query, 1)
...
# 3. 執行查詢
if not sql:
print("\nSQL生成失敗")
return Result.error()
resultSet = queryService.query_with_column(sql)
...2、LangChain框架
Agent模式初始化傳入tools即可：
# 初始化Agent
agent = initialize_agent(
tools=[semantic_tool, sql_tool],
llm=llm,
agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
verbose=True,
handle_parsing_errors=True,
max_iteratinotallow=10, # 增加最大迭代次數以支持多步查詢
early_stopping_method="generate",
memory=memory
)3、QwenAgent框架
Agent模式初始化傳入tools即可：
# 創建Agent實例
agent = Assistant(
name='ai_agent_assistant',
llm={
'model': 'qwen3:32b',
'model_server': 'http://localhost:11434/v1',
},
system_message="""
你是一個數據分析助手，負責幫助用户查詢數據庫信息。
請特別注意：用户的問題可能需要從多個表中查詢數據。
1. 首先確定需要查詢哪些數據
2. 使用match_metadata工具分別匹配包含這些數據的表結構
3. 對每個表生成相應的SQL查詢語句
4. 執行查詢並彙總結果
5. 最後計算並給出答案
請確保逐步執行，不要跳過任何步驟。
""",
function_list=[MatchMetadataTool(), ExecuteSQLTool()],
)需要處理一下流式輸出結果，獲取文本內容：
def read_steam_response(response_generator):
# 處理生成器響應
full_response = ''
start = 0
end = 0
for response in response_generator:
# 檢查響應類型並適當處理
if isinstance(response, list):
# 如果是列表，提取內容
for item in response:
ifisinstance(item, dict) and'content'in item:
full_response = item['content']
end = full_response.__len__()
elifisinstance(item, str):
full_response = item
end = full_response.__len__()
elif isinstance(response, dict) and'content'in response:
full_response = response['content']
end = full_response.__len__()
elif isinstance(response, str):
full_response = response
end = full_response.__len__()
print(f"{full_response[start:end]}", end="")
start = end
return full_response4、AgentScope框架
Agent模式初始化傳入tools即可：
# 準備工具
toolkit = Toolkit()
toolkit.register_tool_function(match_metadata)
toolkit.register_tool_function(execute_sql)
jarvis = ReActAgent(
name="Jarvis",
sys_prompt="""
你是一個數據分析助手，負責幫助用户查詢數據庫信息。
請特別注意：用户的問題可能需要從多個表中查詢數據。
1. 首先確定需要查詢哪些數據
2. 使用match_metadata工具分別匹配包含這些數據的表結構
3. 對每個表生成相應的SQL查詢語句
4. 執行查詢並彙總結果
5. 最後計算並給出答案
請確保逐步執行，不要跳過任何步驟。
""",
model=OllamaChatModel(
model_name="qwen3:32b", # 指定模型名稱
stream=True, # 根據需要設置是否流式輸出
enable_thinking=True, # 為Qwen3啓用思考功能（可選）
# host="http://localhost:11434" # 如果Ollama不在默認地址，需指定
),
formatter=OllamaChatFormatter(),
toolkit=toolkit,
memory=InMemoryMemory(),
)小結
手寫開發中需顯式調用語義檢索、SQL查詢等工具，幾種框架都支持通過工具註冊機制自動完成調用與路由，降低耦合並增強複用。
上下文記憶對比
在AI Agent實踐過程中，如果需要實現多輪對話，就會碰到上下文記憶的處理，所以接下來對這塊做個對比。
1、手寫代碼
這種模式需要手寫代碼來管理上下文消息：
import threading
from dataclasses import dataclass
from typing importList, Dict, Any
@dataclass
class Message:
role: str# "system", "user", "assistant"
content: str
def to_dict(self) -> Dict[str, str]:
return {"role": self.role, "content": self.content}
class MessageManager:
def __init__(self, max_history: int = 10):
"""
初始化消息管理器
:param max_history: 每個 session 最多保留的歷史消息數量（不包括 system 消息）
"""
self.max_history = max_history
self._sessions: Dict[str, Dict[str, Any]] = {}
self._lock = threading.Lock() # 保證線程安全
def set_system_message(self, session_id: str, content: str) -> None:
"""為指定 session 設置 system 消息（會覆蓋舊的）"""
with self._lock:
if session_id notinself._sessions:
self._sessions[session_id] = {
"system": None,
"history": [] # 只存 user/assistant 對話
}
self._sessions[session_id]["system"] = Message(role="system", cnotallow=content)
def add_user_message(self, session_id: str, content: str) -> None:
"""添加用户消息"""
self._add_message(session_id, "user", content)
def add_assistant_message(self, session_id: str, content: str) -> None:
"""添加助手回覆"""
self._add_message(session_id, "assistant", content)
def _add_message(self, session_id: str, role: str, content: str) -> None:
with self._lock:
if session_id notin self._sessions:
self._sessions[session_id] = {
"system": None,
"history": []
}
history = self._sessions[session_id]["history"]
history.append(Message(role=role, cnotallow=content))
# 限制歷史長度（只保留最近的 max_history 條 user/assistant 消息）
if len(history) > self.max_history * 2: # 每輪對話含 user + assistant
# 保留最後 max_history * 2 條
self._sessions[session_id]["history"] = history[-(self.max_history * 2):]
def get_messages(self, session_id: str) -> List[Dict[str, str]]:
"""獲取可用於 Ollama /api/chat 的 messages 列表"""
with self._lock:
session = self._sessions.get(session_id)
if not session:
return []
messages = []
# 添加 system 消息（如果有）
if session["system"]:
messages.append(session["system"].to_dict())
# 添加歷史對話
for msg in session["history"]:
messages.append(msg.to_dict())
return messages
def clear_session(self, session_id: str) -> None:
"""清除指定 session 的所有消息"""
with self._lock:
self._sessions.pop(session_id, None)
def list_sessions(self) -> List[str]:
"""列出所有 session ID"""
with self._lock:
returnlist(self._sessions.keys())
def delete_session(self, session_id: str) -> bool:
"""刪除 session，返回是否刪除成功"""
with self._lock:
if session_id inself._sessions:
del self._sessions[session_id]
return True
return False調用時手動傳入：
# sql agent
def sql_agent(user_query):
# 1. 語義匹配
table = semanticService.hybrid_search(user_query, 1)
if not table:
print(f"未匹配到字段")
return Result.error()
table_struct = [t["table_info"] for t in table]
prompt = f"""
你是一個MySQL專家。根據以下表結構信息：
{table_struct}
歷史問答（僅供參考）："{mm.get_messages(session_id)}"
用户查詢："{user_query}"
生成標準MYSQL查詢語句。
要求：
1. 只輸出MYSQL語句，不要額外解釋
2. 根據語義和字段類型，使用COUNT/SUM/AVG等聚合函數進行計算，非必須
3. 給生成的字段取一個簡短的中文名稱
輸出格式：使用[]包含sql文本即可，不需要其他輸出，便於解析，例如:[select 1 from dual]
"""
print(f"SQL AGENT PROMPT={prompt}")
# 2. 大模型生成SQL
str1 = analysisService.analysis(prompt)
sql = re.search(r'\[(.*?)\]', str1, re.DOTALL).group(1).strip()
# 3. 執行查詢
if not sql:
print("\nSQL生成失敗")
return Result.error()
resultSet = queryService.query_with_column(sql)
if not resultSet:
print("\nSQL查詢失敗")
return Result.error()
return Result.success(data={ "tableStruct": table_struct, "resultSet": resultSet, "sql": sql })
# analysis agent
def analysis_agent(user_query, data):
# 基礎分析
prompt = f"""
根據以下表結構信息：
{data['tableStruct']}
查詢SQL：
{data['sql']}
和以下數據信息：
{data['resultSet']}
歷史問答（僅供參考）："{mm.get_messages(session_id)}"
用户查詢："{user_query}"
生成一段簡要分析，加上一些預測總結的內容
"""
print(f"ANALYSIS AGENT PROMPT={prompt}")
return Result.success(analysisService.analysis(prompt))2、LangChain框架
Agent模式指定：
# 記憶
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)
# 初始化Agent
agent = initialize_agent(
tools=[semantic_tool, sql_tool],
llm=llm,
agent=AgentType.CONVERSATIONAL_REACT_DESCRIPTION,
verbose=True,
handle_parsing_errors=True,
max_iteratinotallow=10, # 增加最大迭代次數以支持多步查詢
early_stopping_method="generate",
memory=memory
)除了上面用到的 ConversationBufferMemory，LangChain還提供了多種Memory組件，以適應不同場景的需求。可以根據具體情況選擇：
注意確認Agent類型兼容性：在選用Memory前，請確認你使用的AgentType 支持Memory功能。例如，ZERO_SHOT_REACT_DESCRIPTION 默認不支持記憶，而CONVERSATIONAL_REACT_DESCRIPTION則專為多輪對話設計。
注意上下文長度：為Agent添加記憶會佔用模型的上下文窗口。如果對話很長，考慮使用Conversation Summary Memory或Conversation Buffer Window Memory來避免超出限制。
調試工具：將 verbose 參數設為 True，可以在控制枱看到詳細的決策過程，有助於觀察Memory是否正常工作。
3、QwenAgent框架
Agent模式的Assistant類不支持，需要手動管理：
def analysis(user_query):
try:
# 運行Agent
messages = mm.get_messages(session_id) + [{'role': 'user', 'content': user_query}]
response_generator = agent.run(messages=messages)
# 處理生成器響應
full_response = ''
start = 0
end = 0
for response in response_generator:
# 檢查響應類型並適當處理
ifisinstance(response, list):
# 如果是列表，提取內容
for item in response:
ifisinstance(item, dict) and'content'in item:
full_response = item['content']
end = full_response.__len__()
elifisinstance(item, str):
full_response = item
end = full_response.__len__()
elifisinstance(response, dict) and'content'in response:
full_response = response['content']
end = full_response.__len__()
elifisinstance(response, str):
full_response = response
end = full_response.__len__()
print(f"{full_response[start:end]}", end="")
start = end
print(f"最終結果: {full_response}")
# 緩存歷史對話
if full_response:
mm.add_user_message(session_id, user_query)
mm.add_assistant_message(session_id, full_response)
return full_response
except Exception as e:
print(f"執行過程中出錯: {str(e)}")
# 這裏可以添加重試或更詳細的錯誤處理邏輯
return f"錯誤: {str(e)}"4、AgentScope框架
Agent初始化指定對應類型即可：
jarvis = ReActAgent(
name="Jarvis",
sys_prompt="""
你是一個數據分析助手，負責幫助用户查詢數據庫信息。
請特別注意：用户的問題可能需要從多個表中查詢數據。
1. 首先確定需要查詢哪些數據
2. 使用match_metadata工具分別匹配包含這些數據的表結構
3. 對每個表生成相應的SQL查詢語句
4. 執行查詢並彙總結果
5. 最後計算並給出答案
請確保逐步執行，不要跳過任何步驟。
""",
model=OllamaChatModel(
model_name="qwen3:32b", # 指定模型名稱
stream=True, # 根據需要設置是否流式輸出
enable_thinking=True, # 為Qwen3啓用思考功能（可選）
# host="http://localhost:11434" # 如果Ollama不在默認地址，需指定
),
formatter=OllamaChatFormatter(),
toolkit=toolkit,
memory=InMemoryMemory(),
)小結
手寫代碼需自行實現消息管理和上下文傳遞，工作量大且需要手動維護。框架提供多樣化記憶組件，支持對話上下文的靈活管理，適配不同對話場景，有效提升對話質量。
-
多智能體協作對比 -
自從Google提出A2A協議，多智能體協作已經成為業界實踐過程的熱點，為進一步對比手寫代碼和開發框架在多Agent協作上的區別，我們將本案例的程序設計拆分為以下兩個Agent：
1、SQL Agent
- 調用外部語義檢索(RAG)API匹配元數據
- 調用大模型根據元數據和用户提問生成查數SQL
- 調用查數API查詢數據
2、Analysis Agent
- 調用大模型分析數據回答問題
1、手寫代碼
需要手動串聯多個Agent調用，消息格式自定義傳遞：
...
def workflow(user_input):
# 1 - SQL Agent
result = sql_agent(user_input)
if not result.success:
return None
# 2 - Analysis Agent
return analysis_agent(user_input, result.data)
...2、LangChain框架
原生僅支持單體Agent，多Agent需要手動串聯，實現類似手寫代碼。
3、QwenAgent框架
原生僅支持單體Agent，多Agent需要手動串聯，實現類似同手寫代碼。
4、AgentScope框架
基於AgentScope框架，我們可以把 SQLAgent 和 AnalysisAgent 封裝為 兩個智能體（均繼承 AgentBase），然後兩個智能體的協作使用順序管道sequential_pipeline來串聯，實現“多智能體任務流轉”。
AgentScope 中的主要管道類型包括：
這幾種管道協作模式，具體實踐敬請期待下篇文章分享。
小結
手寫代碼多為線性調用，缺乏並行和複雜調度能力。AgentScope框架支持多Agent並行、條件分支、消息廣播等複雜管道模式，適合構建複雜的多智能體協作系統。其他框架都需要手動組合實現。
- 對比總結
-
通過以上對比來看，在日常智能體開發中，尤其是需要多輪對話、實現多工具調用和多智能體協作時，幾種開發框架確實提供了極大的便利和擴展能力，大幅降低開發門檻和維護成本，而手寫代碼則適合高度定製化需求，靈活性最高。
選型建議
🔹 LangChain：適合你搭建任何單體智能體系統（RAG、SQL、工具調用）。
🔹 Qwen-Agent：適合快速構建“基於 Qwen 模型”的問答或助手。
🔹 AgentScope：適合研究或實現多智能體協作系統（Planner、Coder、Critic 等角色聯動）。
🔹 手寫代碼：高度定製化開發，當我們的手寫代碼逐步抽象完善，使用的頻率多了，就成了框架！
