AI Agent在投资领域比较火爆，代码助手的投资也比较多，单AI Agent，多角色AI Agent逐渐会成为AI应用的方向。最近一个10岁孩子直接使用豆包和两个ChatGPT进行愚公该移山还是该搬家的辩论活动吸引了大家的眼球，用户作为主持人角色指导2个GPT进行正反方辩友进行辩论和1个GPT作为评委进行评价，这个就是典型的多角色AI代理的应用场景，使用了3个GPT，其实一个也可以的，虽然这个活动最终没有达到预期，Chat GPT作为评委的代理，虽然没有评论，但都记录了它“听”到的内容。这个场景是直接使用GPT作为代理的例子，多角色AI Agent打开走向AGI的想象空间。下面以我比较熟悉的领域，尝试实现一个多角色AI代理编写代码的例子。涉及的角色为：
用户UserInterface：输入任务指令，并在任务无法完成时，进行指导，如：修改代码。
需求分析RequirementAnalyzer：针对用户输入的指令进行需求分析。
代码结构ProgramStructure：针对需求，提出合理的代码结构。
编写代码CodeGenerator：根据需求及代码结构和语言设置，编写代码。
代码审查CodeReviewer：针对生成的代码进行代码审查，提出改进意见。
代码执行CodeExecutor：执行审查完成的代码，输出测试结果。
协作交互Coordinator：协作交互逻辑，通过多次迭代完成任务。
本文默认使用的ollama版本的qwen2.5 7b模型。
完整代码：
import gradio as gr
import requests
import subprocess
import logging
import json
from io import StringIO
import threading
import time
配置日志记录
log_stream = StringIO()
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(), # 输出到控制台
logging.StreamHandler(log_stream) # 输出到 StringIO 流
]
)
定义RequirementAnalyzer角色
class RequirementAnalyzer:
def init(self, api_url, model):
self.api_url = api_url
self.model = model
def analyze_instruction(self, instruction):
messages = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": f"分析以下指令: {instruction}"}
]
response = requests.post(self.api_url, json={"messages": messages, "model": self.model, "stream": True}, stream=True)
analysis = self.process_stream(response)
logging.info(f"[RequirementAnalyzer] 分析的需求: {analysis}")
return analysis
def process_stream(self, response):
content = ""
for chunk in response.iter_content(chunk_size=None):
if chunk:
chunk_data = chunk.decode('utf-8')
if "done" in chunk_data:
chunk_data = chunk_data.split("data: ")[-1]
chunk_json = json.loads(chunk_data)
if chunk_json["done"]:
break
else:
content += chunk_json["message"]["content"]
return content
定义ProgramStructure角色
class ProgramStructure:
def init(self, api_url, model):
self.api_url = api_url
self.model = model
def determine_structure(self, requirement):
messages = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": f"确定以下需求的类结构: {requirement}"}
]
response = requests.post(self.api_url, json={"messages": messages, "model": self.model, "stream": True}, stream=True)
structure = self.process_stream(response)
logging.info(f"[ProgramStructure] 确定的结构: {structure}")
return structure
def process_stream(self, response):
content = ""
for chunk in response.iter_content(chunk_size=None):
if chunk:
chunk_data = chunk.decode('utf-8')
if "done" in chunk_data:
chunk_data = chunk_data.split("data: ")[-1]
chunk_json = json.loads(chunk_data)
if chunk_json["done"]:
break
else:
content += chunk_json["message"]["content"]
return content
定义CodeGenerator角色
class CodeGenerator:
def init(self, api_url, model):
self.api_url = api_url
self.model = model
def generate_code(self, requirement, structure, language):
messages = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": f"根据以下需求和结构编写{language}代码: {requirement}, {structure}"}
]
response = requests.post(self.api_url, json={"messages": messages, "model": self.model, "stream": True}, stream=True)
code = self.process_stream(response)
code = self.extract_code(code) # 提取代码部分
logging.info(f"[CodeGenerator] 生成的代码: {code}")
return code
def process_stream(self, response):
content = ""
for chunk in response.iter_content(chunk_size=None):
if chunk:
chunk_data = chunk.decode('utf-8')
if "done" in chunk_data:
chunk_data = chunk_data.split("data: ")[-1]
chunk_json = json.loads(chunk_data)
if chunk_json["done"]:
break
else:
content += chunk_json["message"]["content"]
return content
def extract_code(self, content):
# 假设代码部分被包裹在 ``` 代码块中，并去掉打头的语言标识
import re
match = re.search(r'```.*?\n(.*?)```', content, re.DOTALL)
if match:
return match.group(1).strip()
return content.strip()
定义CodeReviewer角色
class CodeReviewer:
def init(self, api_url, model):
self.api_url = api_url
self.model = model
def review_code(self, code, language):
messages = [
{"role": "system", "content": "你是一个有帮助的助手。"},
{"role": "user", "content": f"审查以下{language}代码并提出改进建议，并给出结论，如：整体来说，这段代码已经很完善: {code}"}
]
response = requests.post(self.api_url, json={"messages": messages, "model": self.model, "stream": True}, stream=True)
review = self.process_stream(response)
logging.info(f"[CodeReviewer] 代码审查: {review}")
return review
def process_stream(self, response):
content = ""
for chunk in response.iter_content(chunk_size=None):
if chunk:
chunk_data = chunk.decode('utf-8')
if "done" in chunk_data:
chunk_data = chunk_data.split("data: ")[-1]
chunk_json = json.loads(chunk_data)
if chunk_json["done"]:
break
else:
content += chunk_json["message"]["content"]
return content
定义CodeExecutor角色
class CodeExecutor:
def execute_code(self, code, language):
if language == "Python":
with open("temp.py", "w") as f:
f.write(code)
try:
result = subprocess.run(["python", "temp.py"], capture_output=True, text=True)
logging.info(f"[CodeExecutor] 执行结果: {result.stdout}")
return result.stdout
except Exception as e:
logging.error(f"[CodeExecutor] 执行错误: {e}")
return str(e)
elif language == "JavaScript":
with open("temp.js", "w") as f:
f.write(code)
try:
result = subprocess.run(["node", "temp.js"], capture_output=True, text=True)
logging.info(f"[CodeExecutor] 执行结果: {result.stdout}")
return result.stdout
except Exception as e:
logging.error(f"[CodeExecutor] 执行错误: {e}")
return str(e)
else:
logging.error(f"[CodeExecutor] 不支持的语言")
return "不支持的语言"
定义UserInterface角色
class UserInterface:
def init(self, coordinator):
self.coordinator = coordinator
self.log_output = None
def ai_code_assistant(self, instruction, language, max_attempts):
return self.coordinator.process_instruction(instruction, language, max_attempts)
def update_logs(self):
while True:
time.sleep(1)
if self.log_output:
self.log_output.update(value=log_stream.getvalue())
def launch(self):
with gr.Blocks() as iface:
gr.Markdown("# AI代码助手")
with gr.Row():
with gr.Column():
instruction = gr.Textbox(label="指令")
language = gr.Dropdown(["Python", "JavaScript"], label="语言")
max_attempts = gr.Number(label="最大交互次数", value=5)
submit_btn = gr.Button("提交")
with gr.Column():
with gr.Tabs():
with gr.TabItem("生成的代码"):
generated_code = gr.Textbox(label="生成的代码", lines=10)
with gr.TabItem("程序结构"):
program_structure = gr.Textbox(label="程序结构", lines=10)
with gr.TabItem("审查结果"):
review_result = gr.Textbox(label="审查结果", lines=10)
with gr.TabItem("执行结果"):
execution_result = gr.Textbox(label="执行结果", lines=10)
with gr.TabItem("日志输出"):
self.log_output = gr.Textbox(label="日志输出", lines=10)
with gr.TabItem("修改代码"):
modified_code = gr.Textbox(label="修改代码", lines=10)
modify_btn = gr.Button("提交修改")
submit_btn.click(
fn=self.ai_code_assistant,
inputs=[instruction, language, max_attempts],
outputs=[generated_code, program_structure, review_result, execution_result, self.log_output]
)
modify_btn.click(
fn=self.coordinator.process_modified_code,
inputs=[modified_code, language],
outputs=[generated_code, program_structure, review_result, execution_result, self.log_output]
)
gr.Examples(
examples=[
["计算一个数的阶乘", "Python", 5],
["对一个数字列表按升序排序", "JavaScript", 5]
],
inputs=[instruction, language, max_attempts],
outputs=[generated_code, program_structure, review_result, execution_result, self.log_output]
)
# 启动日志更新线程
threading.Thread(target=self.update_logs, daemon=True).start()
iface.launch()
定义Coordinator角色
class Coordinator:
def init(self, requirement_analyzer, program_structure, code_generator, code_reviewer, code_executor):
self.requirement_analyzer = requirement_analyzer
self.program_structure = program_structure
self.code_generator = code_generator
self.code_reviewer = code_reviewer
self.code_executor = code_executor
self.interaction_count = 0
self.max_attempts = 5
self.current_code = None
self.current_language = None
self.continue_after_max_attempts = False
def process_instruction(self, instruction, language, max_attempts):
self.max_attempts = max_attempts
self.current_language = language
self.continue_after_max_attempts = False
# 分析用户指令
requirement = self.requirement_analyzer.analyze_instruction(instruction)
# 确定程序结构
structure = self.program_structure.determine_structure(requirement)
# 循环生成和审查代码，直到审查通过或达到最大交互次数
while self.interaction_count < self.max_attempts or self.continue_after_max_attempts:
self.interaction_count += 1
logging.info(f"[Coordinator] 第 {self.interaction_count} 次尝试")
# 生成代码
code = self.code_generator.generate_code(requirement, structure, language)
self.current_code = code
# 审查代码
review = self.code_reviewer.review_code(code, language)
# 如果审查通过，跳出循环
if "代码已经很完善" in review:
break
# 否则，重新生成代码
requirement = f"修复以下代码: {code}"
# 如果达到最大交互次数，提示用户修改代码
if self.interaction_count >= self.max_attempts and not self.continue_after_max_attempts:
logging.info("[Coordinator] 已达到最大交互次数，任务结束。")
return code, structure, review, "已达到最大交互次数，任务结束。", self.get_logs()
# 执行代码
result = self.code_executor.execute_code(code, language)
# 检查执行结果，如果发现问题，驱动相关角色行动
if "Error" in result:
self.handle_error(code, result)
return code, structure, review, result, self.get_logs()
def handle_error(self, code, error_message):
# 这里可以实现错误处理逻辑，例如重新生成代码或提示用户修改代码
logging.error(f"[Coordinator] 检测到错误: {error_message}")
logging.info("[Coordinator] 驱动相关角色采取行动...")
# 提示用户修改代码
self.current_code = code
return code, "", "", f"检测到错误，请修改代码: {error_message}", self.get_logs()
def process_modified_code(self, modified_code, language):
self.current_code = modified_code
self.continue_after_max_attempts = True
# 执行修改后的代码
result = self.code_executor.execute_code(modified_code, language)
# 检查执行结果，如果发现问题，驱动相关角色行动
if "Error" in result:
self.handle_error(modified_code, result)
return modified_code, "", "", result, self.get_logs()
def get_logs(self):
# 获取日志输出
return log_stream.getvalue()
主程序
def main():
# 初始化API URL和模型
api_url = "http://localhost:11434/api/chat"
model = "qwen2.5"
# 创建角色实例
requirement_analyzer = RequirementAnalyzer(api_url, model)
program_structure = ProgramStructure(api_url, model)
code_generator = CodeGenerator(api_url, model)
code_reviewer = CodeReviewer(api_url, model)
code_executor = CodeExecutor()
coordinator = Coordinator(requirement_analyzer, program_structure, code_generator, code_reviewer, code_executor)
user_interface = UserInterface(coordinator)
# 启动用户界面
user_interface.launch()
if name == "main":
main()
首次发表地址：https://blog.csdn.net/wxl781227/article/details/143475737
