python机器人Agent编程——使用swarm框架和ollama实现一个本地大模型和爬虫结合的手机号归属地天气查询Agent流（体会）

🕗 发布于 2024-11-10 09:23 python 机器人爬虫

一、前言

前一篇博文，我们使用qwen_agent库，及fastapi服务来实现了一个手机号归属地天气查询工具，运行发现一个问题，就是大部分运行时，系统时无法实现多个agent的调用的，很多情况下一次查询只能调用一个function。搜索了一些相关资料，发现，目前很多多agent框架也没有很好的解决这样的问题。个人判断，目前这项技术应该还没有成熟，或者开源的框架还没有很好的表现。比如crewai等。最佳openai开源了swarm框架，试了一下，也没找到一个很好的办法。本篇，实现的是固定的流程，就是手动设置几个agent处理消息的流程，主要是串行的流程，来实现输入手机号码，三个上下游的agent像流水线一样，进行相应的信息处理包括调用function，最后返回城市及天气信息：
在这里插入图片描述

二、代码实现

2.1关于本地模型设置

这里需要注意的是，要实现swarm调用ollama本地模型，本人测试，大模型版本必须是qwen需要是2.5，和llama必须是3.2，低于这个版本，会报不能调用tool功能的错误。swarm默认是openai的自家的模型，因此这里会存在兼容性的问题，如果用到生产环境，不排除可能会有硬伤。
要实现ollama本地模型，参考了网上的办法，需要做如下设置：

  
from openai import OpenAI
# 定义本地模型  
MODEL = "llama3.2:latest"  
#MODEL = "qwen2.5:0.5b"
ollama_client = OpenAI(
    base_url = 'http://localhost:11434/v1',
    api_key='None', # required, but unused
)
# 初始化 Swarm 客户端用本地模型 
client = Swarm(client=ollama_client)

2.2智能体的定义

智能体定义比较简洁，这个非常直观，需要注意的是，如果是使用本地大模型，每个agnet定义时，都要指定模型的名称，例如：


  
# 创建1个智能体  
search_agent = Agent(  
    name="receptionist agent",  
    instructions="""  
    You are a receptionist, and you need to respond according to the user's requests. Your tasks are: 
    1) If the user provides a phone number and does not ask about the weather, you should call the get_mobile_address function and then directly return the query result; 
    2) If the user provides a phone number and requests to check the weather, you should first call the mobile phone origin query tool, then call the weather query tool, and finally return the query results.

    """,  
    functions=[get_mobile_address],  #这里定义这个智能体可以使用的function
    model=MODEL  #这里需要指定前面定义的模型
)

2.3定义调用的function

智能体是可以调用function（也叫tool？）的，swarm的function定义相较于qwen-agent和lanchain等框架，现得非常轻量，就是常规的function格式就ok，他会主动帮你生成json schema。

# 定义手机号归属地搜索 Function，并返回查询结果  
def get_mobile_address(number:str):  
    """Search for the location according the number input"""     
    res=p.find(number)
    print("1原始查询结果:",type(res),res)
    if res:
        if "city" in res.keys(): 
            key=  res['city']    
            if key in citycode:                
                code=citycode[key]       
                print("城市的代码：",code)
                return f"according the phon number:{number},it belong to the city whose name is:{key},according the name of city I find the citynumber,the citynumber is "+code
            else:
                print("未找到城市的代码")
                return "sorry I cant find the citynumber"            
            
        else:
            print("未找到手机号归属地，抱歉")
            return "sorry I cant find the citynumber"
    else:
        print("未找到手机号归属地，抱歉")
        return "未找到手机号归属地，抱歉"

2.4定义Agent的流

这里可以设置一个Agent先后调用的串行流程，本例流程是用户输入了一个电话号码，接下来第一个agent拿到用户的输入，就根据用户的输入是否含有电话号码，如果有就调用一个funciton查归属地，如果没有就不调用（这里发现无论输没输电话号码都会调用function,估计跟大模型的智商有关系，估计好的模型应该不会这么傻叉，待验证)。第一个Agent完成任务后，就交给第二个Agent，第二个Agent根据第一个完成情况，进行查询天气或者直接返回无法查询，然后交给第三个Agent，这个Agent负责接收第二个的结果信息，进行总结提炼，然后形成一个报告返回给用户，流程结束。

# 工作流程，按顺序处理
def process_ask(topic):  
    """Run the searching workflow"""  
    with st.spinner("Processing ..."):  # Changed to st.spinner for better context
        # Search  
        st.write(f"1.Searching the location for your phone number...{topic}")  
        search_response = client.run(  
            agent=search_agent,  
            messages=[{"role": "user", "content": f"what is location of this number:{topic}"}]  
        )  
        response1 = search_response.messages[-2]["content"] 
        print("search_response.messages\n",response1)
        # Synthesize  
        st.write(f"2.weather_agent processing...")  
        synthesis_response = client.run(  
            agent=weather_agent,  
            messages=[{"role": "user", "content": f"{response1}"}]  
        )  
        response2 = synthesis_response.messages[-1]["content"]  
        print("synthesis_response.messages\n",response2)
        # Summarize  
        st.write(f"3.Creating summary...")  
        summary_response = client.run(  
            agent=summary_agent,  
            messages=[{"role": "user", "content": f"Summarize this information:\n{response2}"}]  
        )  
     
        return summary_response.messages[-1]["content"]

2.5发布到streamlit的交互页上

加上一层web客户端，用streamlit框架，接收用户输入，返回流程进度及查询结果信息：

# 用户web交互界面  
topic = st.text_input("Enter your number:", value="13858130853")  
if st.button("Process searching", type="primary"):  
    if topic:  
        try:  
            final_summary = process_ask(topic)  
            st.header(f"Searching: {topic}")  
            st.markdown(final_summary)  
        except Exception as e:  
            st.error(f"An error occurred: {str(e)}")  
    else:  
        st.error("Please enter a number!")

三、完整代码

完整代码如下：

import streamlit as st  
from swarm import Swarm, Agent  

import requests
import re
from phone import Phone
from city import *
  
from openai import OpenAI
# 定义模型  
MODEL = "llama3.2:latest"  
#MODEL = "qwen2.5:0.5b"
ollama_client = OpenAI(
    base_url = 'http://localhost:11434/v1',
    api_key='None', # required, but unused
)
# 初始化 Swarm 客户端  
client = Swarm(client=ollama_client)  
  
# 通过 Streamlit 创建用户界面，为页面和应用程序添加一个标题  
st.set_page_config(page_title="AI News Processor", page_icon="📰")  
st.title("📰 Phone Location And Weather Searcher Agent")  

p = Phone()  
# 定义新闻搜索 Function，使用 DuckDuckGo 搜索 API，获取当月新闻，并返回结构化结果  
def get_mobile_address(number:str):  
    """Search for the location according the number input"""     
    res=p.find(number)
    print("1原始查询结果:",type(res),res)
    if res:
        if "city" in res.keys(): 
            key=  res['city']    
            if key in citycode:                
                code=citycode[key]       
                print("城市的代码：",code)
                return f"according the phon number:{number},it belong to the city whose name is:{key},according the name of city I find the citynumber,the citynumber is "+code
            else:
                print("未找到城市的代码")
                return "sorry I cant find the citynumber"            
            
        else:
            print("未找到手机号归属地，抱歉")
            return "sorry I cant find the citynumber"
    else:
        print("未找到手机号归属地，抱歉")
        return "未找到手机号归属地，抱歉"


def get_weather(citynumber: str):
    """Search for the weather according the city name"""     
    print("2 输入查询的天气对应城市号码：",citynumber)
    try:
        url=f'https://www.weather.com.cn/weather1d/{citynumber}.shtml#input'         
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36 Edg/130.0.0.0",
            "Cookie": "userNewsPort0=1; defaultCty=101010100; defaultCtyName=%u5317%u4EAC; f_city=%E5%8C%97%E4%BA%AC%7C101010100%7C; Hm_lvt_080dabacb001ad3dc8b9b9049b36d43b=1730788105,1730874362; HMACCOUNT=5E07A605B65ACF5F; Hm_lpvt_080dabacb001ad3dc8b9b9049b36d43b=1730875029"
        }
        response = requests.get(url, headers=headers)
        if response.status_code == 200:
            html_content = response.content.decode('utf-8')  # Decode response content to support Chinese
            #print(html_content)
            # 使用正则表达式查找<script>标签中的变量
            script_matches = re.findall(r'<script.*?>(.*?)</script>', html_content, re.DOTALL)
            match=False
            if len(script_matches)>4:
                hour3data = script_matches[4] 
                import json
                
                # Extract the hour3data variable from the script
                hour3data_str = hour3data
                
                # Parse the string to extract the 7d data
                hour3data_json = json.loads(hour3data_str.split('=')[1].strip().rstrip(';'))
                seven_day_data = hour3data_json["7d"]
                
                # Format the extracted data for output
                formatted_data = []
                for day in seven_day_data:
                    formatted_data.append(", ".join(day))
                
                # Join the formatted data into a single string
                hour3data_output = "\n".join(formatted_data)
                print("提取的7天天气数据:", hour3data_output)  
                from datetime import datetime
                # Get the current date and time
                current_datetime = datetime.now()
                current_date_hour = current_datetime.strftime("%Y年-%m月-%d日 %H:00时")
                print("当前日期到小时:", current_date_hour)
                return f"今天的日期和小时时间是 {current_date_hour},the next hours of today's weather and the next 7 day's weather datas is follows:"+hour3data
            else:
                print("未找到天气小时数据")
                return "未找到天气小时数据"
        else:
            print("请求失败，状态码")
            return "请求失败，状态码: {}".format(response.status_code)        
      
    except : 
        print("error")       
        return "error"


  
# 创建智能体  
search_agent = Agent(  
    name="receptionist agent",  
    instructions="""  
    You are a receptionist, and you need to respond according to the user's requests. Your tasks are: 
    1) If the user provides a phone number and does not ask about the weather, you should call the get_mobile_address function and then directly return the query result; 
    2) If the user provides a phone number and requests to check the weather, you should first call the mobile phone origin query tool, then call the weather query tool, and finally return the query results.

    """,  
    functions=[get_mobile_address],  
    model=MODEL  
)  


weather_agent = Agent(  
    name="weather search agent",  
    instructions="""  
    you are a weather search server,your task are:
    1) if the user find a citynumber (which begin with "10" and must be a number), you should use the value of citynumbe(which begin with "10" and must be a number) as parameter and call the get_weather funciton to get the weather information.
    2) if the user did not find a citynumber,do not call any funciton and just respone the orignal message.
    """,  
    functions=[get_weather],  
    model=MODEL  
)  
  
summary_agent = Agent(  
    name="Summarizer weather reporter agent",  
    instructions="""  
    You are a summarization weather reporter, who can analyzes the input information. 
    If there is weather information, Based on the data from the next few hours and the data for the next 7 days, you summarize the weather for today and the weather trends for the next 7 days , and give a weather report,
    you should strictly using the information and do not say anything else out of the information, at the head of the report,you should say the city name belong to phone number,speak in simple chinese.
.   If there is no weather information, you should say sorry to the user.
    """,      
    model=MODEL  
)  
  
# 实施新闻处理工作流程，按顺序处理，显示进度指标  
def process_ask(topic):  
    """Run the searching workflow"""  
    with st.spinner("Processing ..."):  # Changed to st.spinner for better context
        # Search  
        st.write(f"1.Searching the location for your phone number...{topic}")  
        search_response = client.run(  
            agent=search_agent,  
            messages=[{"role": "user", "content": f"what is location of this number:{topic}"}]  
        )  
        response1 = search_response.messages[-2]["content"] 
        print("search_response.messages\n",response1)
        # Synthesize  
        st.write(f"2.weather_agent processing...")  
        synthesis_response = client.run(  
            agent=weather_agent,  
            messages=[{"role": "user", "content": f"{response1}"}]  
        )  
        response2 = synthesis_response.messages[-1]["content"]  
        print("synthesis_response.messages\n",response2)
        # Summarize  
        st.write(f"3.Creating summary...")  
        summary_response = client.run(  
            agent=summary_agent,  
            messages=[{"role": "user", "content": f"Summarize this information:\n{response2}"}]  
        )  
     
        return summary_response.messages[-1]["content"]  
  
# 用户web交互界面  
topic = st.text_input("Enter your number:", value="13858130853")  
if st.button("Process searching", type="primary"):  
    if topic:  
        try:  
            final_summary = process_ask(topic)  
            st.header(f"Searching: {topic}")  
            st.markdown(final_summary)  
        except Exception as e:  
            st.error(f"An error occurred: {str(e)}")  
    else:  
        st.error("Please enter a number!")

四、总结

经过来回运行调试中发现，有如下体会：

本地大模型质量对于Agent的表现非常关键，同样的llama3.2:latest比qwen2.5:0.5b要表现好
提示词也很关键，要写好instructions，这个需要下功夫，反复调试，最好系统学习一下提示工程
中英文混用，可能会影响模型的表现
总体来说多智能体，开源出来的框架表现应该都不佳，没有真正实现模型的自动流程编排，这可能是某些公司的核心技术，还没有公布透露？网友谁有研究请不吝留言赐教。
多智能体协作确实是未来的方向，希望能够出来更智能的框架和很好支持多智能体的本地大模型。
多智能体是解决具身智能自主决策的解决方案

亲爱的网友，您怎么看？

[------------本篇完-------------]

PS.扩展阅读

————————————————————————————————————————

对于python机器人编程感兴趣的小伙伴，可以进入如下链接阅读相关咨询

ps1.六自由度机器人相关文章资源

(1) 对六自由度机械臂的运动控制及python实现（附源码)
在这里插入图片描述

(2) N轴机械臂的MDH正向建模，及python算法
在这里插入图片描述

在这里插入图片描述

（2）python机器人实战——0到1创建一个自动是色块机器人项目-CSDN直播

(3)博文《我从0开始搭建了一个色块自动抓取机器人，并实现了大模型的接入和语音控制-(上基础篇)》的vrep基础环境
(3)博文《我从0开始搭建了一个色块自动抓取机器人，并实现了大模型的接入和语音控制-(上基础篇)》的vrep基础环境
(4)实现了语音输入+大模型指令解析+机器视觉+机械臂流程打通
在这里插入图片描述

(3)python机器人编程——差速AGV机器、基于视觉和预测控制的循迹、自动行驶（上篇）_agv编程-CSDN博客
(4)python机器人编程——差速AGV机器、基于视觉和预测控制的循迹、自动行驶（下篇）_agv路线规划原则python-CSDN博客
对应python及仿真环境资源：源码链接
在这里插入图片描述

ps3.wifi小车控制相关文章资源

web端配套资源源代码已经上传(竖屏版)，下载地址
仿真配套资源已经上传：下载地址
web端配套资源源代码已经上传(横屏版)，下载地址

原文地址：https://blog.csdn.net/kanbide/article/details/143652534

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：单相锁相环，原理与Matlab实现
下一篇：SSH实验1

Windows电脑本地部署llamafile并接入Qwen大语言模型远程AI对话实战
Windows电脑本地部署llamafile并接入Qwen大语言模型远程AI对话实战
阅读更多2024-11-23
解决vm虚拟机中ctrl键粘滞问题（一直处于按下状态）
虚拟机中点击鼠标左键没有问题，单独按下键盘也没有问题，但是如果按下键盘的同时在按下鼠标左键就会卡住。而且100%稳定复现。发现问题原因：vm版本的问题，我的版本是17.5.0，升级成17.5.2 bu
阅读更多2024-11-23
第十章 JavaScript的应用
10.1.1 JavaScript简介1 简单性JavaScript 设计之初就考虑到了非专业程序员的使用，因此它的语法相对简单，容易上手。2 动态性JavaScript 是一种解释型语言，代码可以在
阅读更多2024-11-23
非root用户安装CUDA
4.下载完成后执行.run文件，需要添加两个参数，一个是--override，用于跳过某些检查或限制；另一个是--toolkitpath，它表示安装路径，必须是自己目录下面的路径，如果是根目录就会安装
阅读更多2024-11-23
Python操作neo4j库py2neo使用之py2neo 删除及事务相关操作（三）
【代码】## Python操作neo4j库py2neo使用之py2neo 删除及事务相关操作（三）
阅读更多2024-11-23
【通俗理解】Jensen不等式与变分分布q(z)在积分计算中的应用
Jensen不等式 #变分分布 #积分计算 #期望 #凸函数 #优化问题 #下界估计 #机器学习。
阅读更多2024-11-23
「Mac玩转仓颉内测版27」基础篇7 - 字符串类型详解
本篇将介绍 Cangjie 中的字符串类型，包括字符串的定义、字面量形式、插值表达、常用操作及应用场景，帮助开发者熟练掌握字符串的使用。
阅读更多2024-11-23
数据科学与SQL：组距分组分析 | 区间分布问题
绝对值分布分析也可以理解为组距分组分析。对于某个指标而言，一个记录对应的指标值的绝对值，肯定落在所有指标值的绝对值的最小值和最大值构成的区间内，根据一定的算法，在把这个区间划分为等距离的几个小区间，，
阅读更多2024-11-23
c++编译报C1004错误的原因及解决办法
可能是你的代码文件最后没有正确的结束符，或者是缺少了一些必要的头文件或语句。默认磁盘驱动器没有足够的空间用于临时文件，需要大约两倍于源文件的空间。编译器到达了源文件尾但未解析构造。计算结果为假的 #i
阅读更多2024-11-23
mysql 去重补全取出重复变量函数和存储过程
mysql 去重补全取出重复变量函数和存储过程
阅读更多2024-11-23