第三届琶洲算法大赛 由广州市人民政府、 中国人工智能学会联合主办， 中国信息通信研究院、 广州市海珠区人民政府、 广州市科学技术局、 广州市工业和信息化局、 广州市政务服务和数据管理局、 人工智能与数字经济广东省实验室（广州）共同承办， 聚焦 AI 大模型、 AIGC 行业热点， 广泛邀请行业龙头企业、 科研机构、 投资机构、 生态合作企业、 行业协会参与出题、发动征集、评审、技术合作及商业落地机会， 预计将吸引超 5000 支队伍参赛。
方案应利用大语言模型的语义理解和函数调用功能，准确解析用户查询，并通过访问相关法律数据库或 API，提供服务，包括解答法律问题、查询案件信息、检索历史案件和分析司法数据。
本文提出了一个基于Qwen的Agent+代码生成方案：基于反馈的编程Agent，通过问题重写、命名实体识别、制定计划、代码生成、反馈调试、答案可控生成等步骤，实现目标。支持自定义接口，通过增加接口可以满足不同的查询需求。
整体思路
- 输入问题
- Rewriter：问题重写
- Preprocessor：命名实体识别（NER）
- Recognizer：意图识别
- 任务 DAG，拓扑排序
- Filter：过滤数据表和工具
- Programmer：编写代码
- Coder：执行代码
- Debugger：修复错误
- Combiner：答案可控生成
- Postprocessor：正则替换
- 输出答案
工具函数
将 RESTful API 转换为以下格式，方法文档用于构建提示词，ToolException 用于引导模型调试生成代码的错误：
registered_tools = list()
registered_tools_map = {
}
domain = "comm.chatglm.cn"
url_prefix = f"https://{domain}/law_api/s1_b"
team_token = os.getenv('TEAM_TOKEN') or ''
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer ' + team_token
}
def register_tool(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
registered_tools.append(func)
registered_tools_map[func.__name__] = func
return wrapper
@register_tool
def get_company_info(
value: Annotated[str, "字段值"],
):
"""
根据 **公司名称**、**公司简称** 或 **公司代码** 查找上市公司基本信息
根据返回结果判断，结果为空说明该公司是非上市公司，请使用 get_company_register 工具查询 CompanyRegister
Args:
value (str): 公司名称、公司简称或公司代码
Returns:
CompanyInfo: dict
Example:
>>> company_info = get_company_info("示例公司名称")
>>> company_info.get('所属行业')
"""
if is_empty(value):
raise ToolException(traceback='', ename='get_company_info 参数错误', evalue='value 不能为空')
if isinstance(value, list) or isinstance(value, dict):
raise ToolException(traceback='', ename='get_company_info 参数错误', evalue='value 需要字符串类型')
# 如果 value 没有中文且长度大于 6，应该是统一社会信用代码
if no_chinese(value) and len(value) > 6:
raise ToolException(traceback='', ename='get_company_info 参数错误', evalue='value 参数错误')
if value == '示例公司名称':
raise ToolException(traceback='', ename='get_company_info 参数错误', evalue='value 参数错误')
url = f"{url_prefix}/get_company_info"
# 公司名称
response = requests.post(url, json={
"query_conds": {
"公司名称": value
},
"need_fields": []
}, headers=headers)
data = response.json()
if not is_empty(data):
return data
# 公司简称
response = requests.post(url, json={
"query_conds": {
"公司简称": value
},
"need_fields": []
}, headers=headers)
data = response.json()
if not is_empty(data):
return data
# 公司代码
response = requests.post(url, json={
"query_conds": {
"公司代码": value
},
"need_fields": []
}, headers=headers)
data = response.json()
return data
过滤数据表和工具
这一步有点类似于 Text-to-SQL 任务中的模式链接（Schema Linking）：
- 13 个数据表和 24 个接口
- 使用全量的数据表和接口太多
- 使用大模型 + 实体关系图过滤
- 得到只与问题关联的数据表和接口
编写、执行、调试代码
- Programmer：编写代码
- Coder：执行代码
- Debugger：调试代码
这个阶段使用专注于编程任务的 Qwen-Coder 模型效果会更好。
答案可控生成
Programmer 提示词
完成每一个步骤，立即使用 print 函数输出该步骤的 output 变量值。格式如下：
print(f"Step { {步骤编号}} output - { {output变量名}}:", output变量值)
确保每个步骤的结果都被打印出来，以便跟踪代码执行过程。
Combiner 提示词
[问题]
{question}[运行代码]
{ executed_code}[运行结果]
{sub_questions_result_desc}[任务]
- 按照 运行代码 的步骤，提取 运行结果 中所有有结果的步骤输出到答案。
...
总结
本文代码已发布于 Qwen-Article 仓库。
